package com.boot.admin.modules.sys.ai.service;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.boot.admin.base.R;
import com.boot.admin.core.service.BaseServiceImpl;
import com.boot.admin.data.domain.BasePage;
import com.boot.admin.modules.sys.ai.mapper.AiConversationMapper;
import com.boot.admin.modules.sys.ai.model.entity.AiConversation;
import com.boot.admin.modules.sys.ai.model.entity.AiMessage;
import com.boot.admin.modules.sys.ai.model.entity.AiRole;
import com.boot.admin.modules.sys.ai.model.query.AiConversationQuery;
import com.boot.admin.modules.sys.ai.model.vo.AiConversationVO;
import com.boot.admin.modules.sys.ai.model.vo.AiMessageVO;
import com.boot.admin.modules.sys.ai.model.vo.ConversationResponseVO;
import com.boot.admin.modules.sys.ai.model.vo.StartConversationVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * AI对话服务 - 支持流式传输
 *
 * @author 李飞洋
 * @date 2025/9/24
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class AiConversationService extends BaseServiceImpl<AiConversationMapper, AiConversation> {

    private final AiRoleService aiRoleService;
    private final AiMessageService aiMessageService;
    private final AiTtsService aiTtsService;
    private final LocalSseLlmService localWebSocketLlmService;
    private final StreamingWebSocketService streamingWebSocketService;

    /**
     * 开始新对话 - 流式响应
     */
    @Transactional(rollbackFor = Exception.class)
    public SseEmitter startConversationStreaming(String userId, StartConversationVO startVO) {
        SseEmitter sseEmitter = new SseEmitter(300000L); // 5分钟超时
        final AtomicBoolean sseCompleted = new AtomicBoolean(false);

        try {
            // 检查角色是否存在
            AiRole role = aiRoleService.getRoleById(startVO.getRoleId());
            if (null == role || !role.isEnabled()) {
                safeSseCompleteWithError(sseEmitter, sseCompleted, new RuntimeException("角色不存在或未启用"));
                return sseEmitter;
            }

            // 创建对话
            AiConversation conversation = new AiConversation();
            conversation.setId(IdUtil.simpleUUID());
            conversation.setUserId(userId);
            conversation.setRoleId(startVO.getRoleId());
            conversation.setTitle(StrUtil.isBlank(startVO.getTitle()) ?
                    "与" + role.getRoleName() + "的对话" : startVO.getTitle());
            conversation.setStatus(1); // 进行中
            conversation.setCreated(LocalDateTime.now());
            conversation.setLastMessageTime(LocalDateTime.now());

            save(conversation);

            // 设置SSE事件处理器
            setupSseEventHandlers(sseEmitter, sseCompleted, conversation.getId());

            // 发送对话创建成功事件
            safeSseSend(sseEmitter, sseCompleted, SseEmitter.event()
                    .name("conversation_created")
                    .data("{\"conversationId\":\"" + conversation.getId() +
                            "\",\"title\":\"" + conversation.getTitle() + "\"}"));

            // 构造默认用户输入
            String defaultUserInput = "请做自我介绍";
            String aiMessageId = IdUtil.simpleUUID();
            StringBuilder aiContentBuilder = new StringBuilder();

            // 异步处理流式AI响应
            CompletableFuture.runAsync(() -> {
                try {
                    localWebSocketLlmService.generateResponseStreaming(
                            role, null, defaultUserInput,
                            createStreamingCallback(sseEmitter, sseCompleted, conversation, aiMessageId,
                                    aiContentBuilder, 1, false)
                    );

                } catch (Exception e) {
                    log.error("开始对话流式处理失败", e);
                    safeSseCompleteWithError(sseEmitter, sseCompleted, e);
                }
            });

        } catch (Exception e) {
            log.error("初始化开始对话流式处理失败", e);
            safeSseCompleteWithError(sseEmitter, sseCompleted, e);
        }

        return sseEmitter;
    }

    /**
     * 发送文本消息 - 流式响应模式
     */
    @Transactional(rollbackFor = Exception.class)
    public SseEmitter sendTextMessageStreaming(String userId, String conversationId, String content, Boolean enableWebSearch) {
        // 检查对话是否存在且属于用户
        AiConversation conversation = getById(conversationId);
        if (null == conversation || !userId.equals(conversation.getUserId())) {
            throw new RuntimeException("对话不存在");
        }

        if (!conversation.isActive()) {
            throw new RuntimeException("对话已结束");
        }

        return processUserMessageStreaming(conversation, content, enableWebSearch);
    }

    /**
     * 处理用户消息的通用逻辑 - 普通模式
     */
    private R<ConversationResponseVO> processUserMessage(AiConversation conversation, String userContent) {
        try {
            // 获取角色信息
            AiRole role = aiRoleService.getRoleById(conversation.getRoleId());
            if (null == role) {
                return R.NG("角色不存在");
            }

            // 获取对话历史
            List<AiMessageVO> history = aiMessageService.getRecentMessages(conversation.getId(), 10);

            // 获取下一个消息序号
            Integer nextIndex = aiMessageService.getNextMessageIndex(conversation.getId());

            // 保存用户消息
            AiMessage userMessage = new AiMessage();
            userMessage.setId(IdUtil.simpleUUID());
            userMessage.setConversationId(conversation.getId());
            userMessage.setSenderType(1); // 用户
            userMessage.setMessageType(1); // 文本
            userMessage.setContent(userContent);
            userMessage.setMessageIndex(nextIndex);
            userMessage.setCreated(LocalDateTime.now());

            aiMessageService.save(userMessage);

            // 调用大模型生成AI回复
            R<String> llmResult = localWebSocketLlmService.generateResponse(role.getSystemPrompt(), history, userContent);
            if (!llmResult.isSuccess()) {
                return R.NG("生成AI回复失败：" + llmResult.getMessage());
            }

            String aiContent = llmResult.getResult();

            // 保存AI消息
            AiMessage aiMessage = new AiMessage();
            aiMessage.setId(IdUtil.simpleUUID());
            aiMessage.setConversationId(conversation.getId());
            aiMessage.setSenderType(2); // AI
            aiMessage.setMessageType(1); // 文本
            aiMessage.setContent(aiContent);
            aiMessage.setAudioUrl(null);
            aiMessage.setAudioDuration(null);
            aiMessage.setMessageIndex(nextIndex + 1);
            aiMessage.setCreated(LocalDateTime.now());

            aiMessageService.save(aiMessage);

            // 更新对话最后消息时间
            conversation.setLastMessageTime(LocalDateTime.now());
            updateById(conversation);

            // 构造响应
            ConversationResponseVO response = new ConversationResponseVO();
            response.setConversationId(conversation.getId());
            response.setContent(aiContent);
            response.setAudioUrl(null);
            response.setAudioDuration(null);
            response.setMessageIndex(nextIndex + 1);

            return R.OK(response);

        } catch (Exception e) {
            log.error("处理用户消息失败", e);
            return R.NG("处理消息失败：" + e.getMessage());
        }
    }

    /**
     * 处理用户消息的流式逻辑 - 修复版本
     */
    private SseEmitter processUserMessageStreaming(AiConversation conversation, String userContent, Boolean enableWebSearch) {
        SseEmitter sseEmitter = new SseEmitter(300000L); // 5分钟超时
        final AtomicBoolean sseCompleted = new AtomicBoolean(false);

        try {
            // 获取角色信息
            AiRole role = aiRoleService.getRoleById(conversation.getRoleId());
            if (null == role) {
                safeSseCompleteWithError(sseEmitter, sseCompleted, new RuntimeException("角色不存在"));
                return sseEmitter;
            }

            // 获取对话历史
            List<AiMessageVO> history = aiMessageService.getRecentMessages(conversation.getId(), 10);

            // 获取下一个消息序号
            Integer nextIndex = aiMessageService.getNextMessageIndex(conversation.getId());

            // 保存用户消息
            String userMessageId = IdUtil.simpleUUID();
            AiMessage userMessage = new AiMessage();
            userMessage.setId(userMessageId);
            userMessage.setConversationId(conversation.getId());
            userMessage.setSenderType(1); // 用户
            userMessage.setMessageType(1); // 文本
            userMessage.setContent(userContent);
            userMessage.setMessageIndex(nextIndex);
            userMessage.setCreated(LocalDateTime.now());

            aiMessageService.save(userMessage);

            String aiMessageId = IdUtil.simpleUUID();
            StringBuilder aiContentBuilder = new StringBuilder();

            // 设置SSE事件处理器
            setupSseEventHandlers(sseEmitter, sseCompleted, conversation.getId());

            // 异步处理LLM流式响应
            CompletableFuture.runAsync(() -> {
                try {
                    // 调用流式LLM服务，传递网络搜索参数
                    localWebSocketLlmService.generateResponseStreaming(
                            role, history, userContent, enableWebSearch,
                            createStreamingCallback(sseEmitter, sseCompleted, conversation, aiMessageId,
                                    aiContentBuilder, nextIndex + 1, true)
                    );

                } catch (Exception e) {
                    log.error("流式处理失败", e);
                    safeSseCompleteWithError(sseEmitter, sseCompleted, e);
                }
            });

        } catch (Exception e) {
            log.error("初始化流式处理失败", e);
            safeSseCompleteWithError(sseEmitter, sseCompleted, e);
        }

        return sseEmitter;
    }

    /**
     * 设置SSE事件处理器
     */
    private void setupSseEventHandlers(SseEmitter sseEmitter, AtomicBoolean sseCompleted, String conversationId) {
        sseEmitter.onTimeout(() -> {
            log.warn("SSE连接超时 - 对话: {}", conversationId);
            sseCompleted.set(true);
        });

        sseEmitter.onCompletion(() -> {
            log.debug("SSE连接正常完成 - 对话: {}", conversationId);
            sseCompleted.set(true);
        });

        sseEmitter.onError((throwable) -> {
            log.warn("SSE连接错误 - 对话: {}, 错误: {}", conversationId, throwable.getMessage());
            sseCompleted.set(true);
        });
    }

    /**
     * 创建流式回调处理器
     */
    private StreamingCallback createStreamingCallback(SseEmitter sseEmitter, AtomicBoolean sseCompleted,
                                                      AiConversation conversation, String aiMessageId,
                                                      StringBuilder aiContentBuilder, Integer messageIndex,
                                                      boolean enableAudio) {
        return new StreamingCallback() {
            @Override
            public void onTextChunk(String chunk) {
                try {
                    // 检查SSE连接状态
                    if (sseCompleted.get()) {
                        log.debug("SSE连接已完成，跳过发送文本chunk");
                        return;
                    }

                    // 累积AI回复内容
                    aiContentBuilder.append(chunk);

                    // 安全地通过SSE发送文本chunk给前端
                    safeSseSend(sseEmitter, sseCompleted, SseEmitter.event()
                            .name("text_chunk")
                            .data("{\"conversationId\":\"" + conversation.getId() +
                                    "\",\"messageId\":\"" + aiMessageId +
                                    "\",\"chunk\":\"" + escapeJson(chunk) + "\"}"));

                    // 异步调用TTS生成语音chunk（如果启用）
                    if (enableAudio) {
                        CompletableFuture.runAsync(() -> {
                            processAudioChunkWithSync(conversation.getId(), aiMessageId, chunk,
                                    aiRoleService.getRoleById(conversation.getRoleId()), aiContentBuilder.toString());
                        });
                    }

                } catch (Exception e) {
                    log.error("处理文本chunk失败", e);
                }
            }

            @Override
            public void onComplete() {
                try {
                    // 检查SSE连接状态
                    if (sseCompleted.get()) {
                        log.debug("SSE连接已完成，跳过完成处理");
                        return;
                    }

                    // 保存完整的AI消息
                    AiMessage aiMessage = new AiMessage();
                    aiMessage.setId(aiMessageId);
                    aiMessage.setConversationId(conversation.getId());
                    aiMessage.setSenderType(2); // AI
                    aiMessage.setMessageType(1); // 文本
                    aiMessage.setContent(aiContentBuilder.toString());
                    aiMessage.setMessageIndex(messageIndex);
                    aiMessage.setCreated(LocalDateTime.now());

                    aiMessageService.save(aiMessage);

                    // 更新对话最后消息时间
                    conversation.setLastMessageTime(LocalDateTime.now());
                    updateById(conversation);

                    // 安全地发送完成信号
                    safeSseSend(sseEmitter, sseCompleted, SseEmitter.event()
                            .name("complete")
                            .data("{\"conversationId\":\"" + conversation.getId() +
                                    "\",\"messageId\":\"" + aiMessageId + "\"}"));

                    // 如果启用音频，发送音频完成信号
                    if (enableAudio) {
                        streamingWebSocketService.sendAudioComplete(conversation.getId(), aiMessageId);
                    }

                    safeSseComplete(sseEmitter, sseCompleted);

                } catch (Exception e) {
                    log.error("完成流式响应失败", e);
                    safeSseCompleteWithError(sseEmitter, sseCompleted, e);
                }
            }

            @Override
            public void onError(String error) {
                try {
                    // 检查SSE连接状态
                    if (sseCompleted.get()) {
                        log.debug("SSE连接已完成，跳过错误处理");
                        return;
                    }

                    // 安全地发送错误信息
                    safeSseSend(sseEmitter, sseCompleted, SseEmitter.event()
                            .name("error")
                            .data("{\"error\":\"" + escapeJson(error) + "\"}"));

                    safeSseCompleteWithError(sseEmitter, sseCompleted, new RuntimeException(error));
                } catch (Exception e) {
                    log.error("发送错误信息失败", e);
                }
            }
        };
    }

    /**
     * 处理音频chunk并保持与文本同步
     */
    private void processAudioChunkWithSync(String conversationId, String messageId, String textChunk,
                                           AiRole role, String fullTextSoFar) {
        try {
            // 对于过短的文本chunk，累积到一定长度再处理
            if (textChunk.trim().length() < 5) {
                log.debug("文本chunk过短，跳过TTS处理: {}", textChunk);
                return;
            }

            // 调用TTS生成语音chunk
            R<byte[]> ttsResult = aiTtsService.generateAudioChunk(textChunk, role);
            if (ttsResult.isSuccess()) {
                byte[] audioChunk = ttsResult.getResult();

                if (audioChunk.length > 0) {
                    // 通过WebSocket发送语音chunk给前端
                    streamingWebSocketService.sendAudioChunk(conversationId, messageId, audioChunk);
                    log.debug("发送音频chunk: {} bytes, 对应文本: {}", audioChunk.length, textChunk);
                }
            } else {
                log.warn("TTS生成音频chunk失败: {}", ttsResult.getMessage());
                // 发送音频错误但不中断整个流程
                streamingWebSocketService.sendError(conversationId, "TTS处理失败: " + ttsResult.getMessage());
            }
        } catch (Exception e) {
            log.error("处理音频chunk失败", e);
            streamingWebSocketService.sendError(conversationId, "音频处理异常: " + e.getMessage());
        }
    }

    /**
     * 安全地发送SSE数据，检查连接状态
     */
    private void safeSseSend(SseEmitter sseEmitter, AtomicBoolean sseCompleted, SseEmitter.SseEventBuilder event) {
        if (sseCompleted.get()) {
            log.debug("SSE连接已完成，跳过发送数据");
            return;
        }

        try {
            sseEmitter.send(event);
        } catch (IllegalStateException e) {
            if (e.getMessage() != null && e.getMessage().contains("has already completed")) {
                log.debug("SSE连接已完成，设置完成标志");
                sseCompleted.set(true);
            } else {
                log.warn("发送SSE数据时发生异常", e);
                sseCompleted.set(true);
            }
        } catch (Exception e) {
            log.error("发送SSE数据失败", e);
            sseCompleted.set(true);
        }
    }

    /**
     * 安全地完成SSE连接
     */
    private void safeSseComplete(SseEmitter sseEmitter, AtomicBoolean sseCompleted) {
        if (sseCompleted.compareAndSet(false, true)) {
            try {
                sseEmitter.complete();
                log.debug("SSE连接成功完成");
            } catch (IllegalStateException e) {
                if (e.getMessage() != null && e.getMessage().contains("has already completed")) {
                    log.debug("SSE连接已经完成");
                } else {
                    log.warn("完成SSE连接时发生错误", e);
                }
            } catch (Exception e) {
                log.error("完成SSE连接失败", e);
            }
        }
    }

    /**
     * 安全地以错误状态完成SSE连接
     */
    private void safeSseCompleteWithError(SseEmitter sseEmitter, AtomicBoolean sseCompleted, Throwable error) {
        if (sseCompleted.compareAndSet(false, true)) {
            try {
                sseEmitter.completeWithError(error);
                log.debug("SSE连接以错误状态完成: {}", error.getMessage());
            } catch (IllegalStateException e) {
                if (e.getMessage() != null && e.getMessage().contains("has already completed")) {
                    log.debug("SSE连接已经完成");
                } else {
                    log.warn("以错误状态完成SSE连接时发生错误", e);
                }
            } catch (Exception e) {
                log.error("以错误状态完成SSE连接失败", e);
            }
        }
    }

    /**
     * JSON字符串转义
     */
    private String escapeJson(String input) {
        if (input == null) {
            return "";
        }
        return input.replace("\\", "\\\\")
                .replace("\"", "\\\"")
                .replace("\n", "\\n")
                .replace("\r", "\\r")
                .replace("\t", "\\t");
    }

    /**
     * 分页查询
     */
    public BasePage<AiConversationVO> queryPage(AiConversationQuery query) {
        Page<AiConversationVO> page = new Page<>(query.getCurrent(), query.getSize());
        List<AiConversationVO> vos = this.baseMapper.queryPage(page, query);
        return new BasePage<>(page.getCurrent(), page.getSize(), page.getTotal(), vos);
    }

    /**
     * 获取对话详情
     */
    public R<AiConversationVO> getConversationById(String userId, String conversationId) {
        AiConversation conversation = getById(conversationId);
        if (null == conversation || !userId.equals(conversation.getUserId())) {
            return R.NG("对话不存在或无权限访问");
        }

        AiConversationVO vo = BeanUtil.toBean(conversation, AiConversationVO.class);

        // 获取角色名称
        AiRole role = aiRoleService.getRoleById(conversation.getRoleId());
        if (null != role) {
            vo.setRoleName(role.getRoleName());
            vo.setAvatarUrl(role.getAvatarUrl());
        }

        return R.OK(vo);
    }

    /**
     * 获取对话列表 - 支持Query条件筛选
     */
    public R<List<AiConversationVO>> getConversationList(AiConversationQuery query) {
        LambdaQueryWrapper<AiConversation> wrapper = Wrappers.lambdaQuery(AiConversation.class);

        // 用户ID必须
        if (StrUtil.isNotBlank(query.getUserId())) {
            wrapper.eq(AiConversation::getUserId, query.getUserId());
        }

        // 可选筛选条件
        if (StrUtil.isNotBlank(query.getRoleId())) {
            wrapper.eq(AiConversation::getRoleId, query.getRoleId());
        }
        if (query.getStatus() != null) {
            wrapper.eq(AiConversation::getStatus, query.getStatus());
        }
        if (StrUtil.isNotBlank(query.getTitle())) {
            wrapper.like(AiConversation::getTitle, query.getTitle());
        }

        wrapper.orderByDesc(AiConversation::getLastMessageTime);

        List<AiConversation> conversations = list(wrapper);
        List<AiConversationVO> vos = BeanUtil.copyToList(conversations, AiConversationVO.class);

        // 设置角色名称
        for (AiConversationVO vo : vos) {
            AiRole role = aiRoleService.getRoleById(vo.getRoleId());
            if (null != role) {
                vo.setRoleName(role.getRoleName());
                vo.setAvatarUrl(role.getAvatarUrl());
            }
        }

        return R.OK(vos);
    }

    /**
     * 获取对话消息列表 - 按对话顺序排序
     */
    public R<List<AiMessageVO>> getConversationMessagesSorted(String userId, String conversationId) {
        // 验证对话权限
        AiConversation conversation = getById(conversationId);
        if (null == conversation || !userId.equals(conversation.getUserId())) {
            return R.NG("对话不存在或无权限访问");
        }

        // 按消息序号升序排序，确保对话顺序正确
        List<AiMessageVO> messages = aiMessageService.getConversationMessages(conversationId);
        return R.OK(messages);
    }

    /**
     * 删除对话（逻辑删除，同时删除对话下的所有消息）
     */
    @Transactional(rollbackFor = Exception.class)
    public R<String> deleteConversation(String userId, String conversationId) {
        // 验证对话权限
        AiConversation conversation = getById(conversationId);
        if (null == conversation) {
            return R.NG("对话不存在");
        }
        if (!userId.equals(conversation.getUserId())) {
            return R.NG("无权限删除此对话");
        }

        // 删除对话
        removeById(conversationId);

        // 删除对话下的所有消息
        LambdaQueryWrapper<AiMessage> wrapper = Wrappers.lambdaQuery(AiMessage.class)
                .eq(AiMessage::getConversationId, conversationId);
        aiMessageService.remove(wrapper);

        log.info("删除对话成功 - conversationId: {}, userId: {}", conversationId, userId);
        return R.OK("删除成功");
    }

    /**
     * 删除消息
     */
    @Transactional(rollbackFor = Exception.class)
    public R<String> deleteMessage(String userId, String messageId) {
        // 获取消息
        AiMessage message = aiMessageService.getById(messageId);
        if (null == message) {
            return R.NG("消息不存在");
        }

        // 验证对话权限
        AiConversation conversation = getById(message.getConversationId());
        if (null == conversation) {
            return R.NG("关联的对话不存在");
        }
        if (!userId.equals(conversation.getUserId())) {
            return R.NG("无权限删除此消息");
        }

        // 删除消息
        aiMessageService.removeById(messageId);

        log.info("删除消息成功 - messageId: {}, conversationId: {}, userId: {}",
                messageId, message.getConversationId(), userId);
        return R.OK("删除成功");
    }

    /**
     * 流式回调接口
     */
    public interface StreamingCallback {
        void onTextChunk(String chunk);
        void onComplete();
        void onError(String error);
    }
}